

#include "EventLoop.hpp"
#include "Logger.hpp"
#include "Channel.hpp"
#include "Poller.hpp"
#include <cstring>
#include <sys/eventfd.h>
#include <unistd.h>

namespace zephyr {

//线程局部变量
__thread EventLoop *t_loopInThisThread = nullptr;

// 默认Poller IO复用接口的超时时间
const int k_pollTimeMs = 10000; // 10s

/**
 * 创建线程之后主线程和子线程谁先运行是不确定的。
 * 通过一个eventfd在线程之间传递数据的好处是多个线程无需上锁就可以实现同步。
 * eventfd支持的最低内核版本为Linux 2.6.27,在2.6.26及之前的版本也可以使用eventfd，但是flags必须设置为0。
 * 函数原型：
 *     #include <sys/eventfd.h>
 *     int eventfd(unsigned int initval, int flags);
 * 参数说明：
 *      initval,初始化计数器的值。
 *      flags, EFD_NONBLOCK,设置socket为非阻塞。
 *             EFD_CLOEXEC，执行fork的时候，在父进程中的描述符会自动关闭，子进程中的描述符保留。
 * 场景：
 *     eventfd可以用于同一个进程之中的线程之间的通信。
 *     eventfd还可以用于同亲缘关系的进程之间的通信。
 *     eventfd用于不同亲缘关系的进程之间通信的话需要把eventfd放在几个进程共享的共享内存中（没有测试过）。
 */
// 创建weakfd 用来唤醒 subreactor 处理新来的channel
int createEventfd() {
    int evtfd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
    if (evtfd < 0) {
        LOG_FATAL("eventfd error: %s.\n", strerror(errno));
    }
    return evtfd;
}

EventLoop::EventLoop()
    : m_looping(false)
    , m_quit(false)
    , m_threadId(CurrentThread::tid())
    , m_poller(Poller::newDefaultPoller(this))
    , m_wakeupfd(createEventfd())
    , m_wakeupChannel(new Channel(this, m_wakeupfd))
    , m_callingPendingFunctors(false) {

    LOG_DEBUG("EventLoop create %p int thread %d.\n", this, m_threadId);
    if (t_loopInThisThread) {
        LOG_FATAL("Another EventLoop %p exists in this thread %d.\n", t_loopInThisThread, m_threadId);
    } else {
        t_loopInThisThread = this;
    }

    m_wakeupChannel->setReadCallback(std::bind(&EventLoop::handleRead, this));
    m_wakeupChannel->enableReading();
}

EventLoop::~EventLoop() {
    m_wakeupChannel->disableAll();
    m_wakeupChannel->remove();
    close(m_wakeupfd);
    t_loopInThisThread = nullptr;
}

void EventLoop::loop() {
    m_looping = true;
    m_quit = false;

    LOG_INFO("EventLoop %p start looping.", this);

    while (!m_quit) {
        m_activeChannels.clear();
        m_pollReturnTime = m_poller->poll(k_pollTimeMs, &m_activeChannels);
        for (auto channel : m_activeChannels) {
            channel->handleEvent(m_pollReturnTime);
        }
        // 执行当前EventLoop所需要处理的回调操作
        /**
         * 执行当前EventLoop事件循环需要处理的回调操作 对于线程数 >=2 的情况 IO线程 mainloop(mainReactor) 主要工作：
         * accept接收连接 => 将accept返回的connfd打包为Channel => TcpServer::newConnection通过轮询将TcpConnection对象分配给subloop处理
         *
         * mainloop调用queueInLoop将回调加入subloop（该回调需要subloop执行 但subloop还在poller_->poll处阻塞） queueInLoop通过wakeup将subloop唤醒
         **/
        doPendingFunctors();
    }
    LOG_INFO("EventLoop %p stop looping.\n", this);
    m_looping = false;
}

/**
 * 退出事件循环
 * 如果不是eventloop当前线程调用quit，需要唤醒eventloop所属线程的eepollwait
 * 
 * muduo通过wakeup()机制 使用eventfd创建的wakeupFd_ notify 使得mainloop和subloop之间能够进行通信
 */
void EventLoop::quit() {
    m_quit = true;
    if (!isInLoopThread()) {
        wakeup();
    }
}

// 在当前loop中执行cb
void EventLoop::runInLoop(Functor cb) {
    if (isInLoopThread()) { // 在EventLoop线程中执行回调
        cb();
    } else { // 不是Eventloop线程
        queueInLoop(cb);
    }
}

// 把cb放入队列中 唤醒loop所在的线程执行cb
void EventLoop::queueInLoop(Functor cb) {
    {
        std::lock_guard<std::mutex> lock(m_mtx);
        m_pendingFunctors.emplace_back(cb);
    }
    /**
     * || callingPendingFunctors 表示当前loop正在执行回调中
     * 但pendingFunctors有加入新的回调，需要向eventfd发送数据，让下一次poll不阻塞
     * 继续执行回调函数
     */
    if (!isInLoopThread() || m_callingPendingFunctors) {
        wakeup();
    }

}

void EventLoop::handleRead() {
    uint64_t one = 1;
    ssize_t n = read(m_wakeupfd, &one, sizeof(one));
    if (n != sizeof(one)) {
        LOG_ERROR("EventLoop handleRead() reads %lu bytes instead of 8.\n", n);
    }
}

void EventLoop::wakeup() {
    uint64_t one = 1;
    ssize_t n = write(m_wakeupfd, &one, sizeof(one));
    if (n != sizeof(one)) {
        LOG_ERROR("EventLoop wakeup() writes %lu bytes instead of 8.\n", n);
    }
}

void EventLoop::updateChannel(Channel *channel) {
    m_poller->updateChannel(channel);
}

void EventLoop::removeChannel(Channel *channel) {
    m_poller->removeChannel(channel);
}

void EventLoop::hasChannel(Channel *channel) {
    m_poller->hasChannel(channel);
}

void EventLoop::doPendingFunctors() {
    std::vector<Functor> functors;
    m_callingPendingFunctors = true;

    {
        std::lock_guard<std::mutex> lock(m_mtx);
        functors.swap(m_pendingFunctors);
    }

    for (const Functor &functor : functors) {
        functor();
    }

    m_callingPendingFunctors = false;
}



}